

package rafthttp

import (
	"context"
	"sync"
	"time"

	"go.etcd.io/etcd/client/pkg/v3/types"
	"go.etcd.io/etcd/raft/v3"
	"go.etcd.io/etcd/raft/v3/raftpb"
	"go.etcd.io/etcd/server/v3/etcdserver/api/snap"
	stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"

	"go.uber.org/zap"
	"golang.org/x/time/rate"
)

const (
	// ConnReadTimeout和ConnWriteTimeout是HTTP pkg创建的每个连接上设置的i/o超时。
	// 5秒超时足以回收坏连接。或者我们必须等待
	// tcp keepalive无法检测到错误连接，这是分钟级别。
	// 对于长期流式连接，RAFT HTTP pkg发送应用程序级别的LinkHeartBeat消息
	// 以保持连接处于活动状态。
	// 对于短期管道连接，必须终止连接以避免其成为
	// 放回http pkg连接池。
	DefaultConnReadTimeout  = 5 * time.Second
	DefaultConnWriteTimeout = 5 * time.Second

	recvBufSize = 4096
	// MaxPendingPropositions在一个领导人选举过程中持有提案。
	// 通常一次领导人选举最多需要1秒。应该有
	// 0-2次选举冲突，每一次需要0.5秒。
	// 我们假设并发提议者的数量小于4096。
	// 一个客户在其提案上阻塞至少1秒，因此4096足够
	// 保存所有提案。
	maxPendingProposals = 4096

	streamAppV2 = "streamMsgAppV2"
	streamMsg   = "streamMsg"
	pipelineMsg = "pipeline"
	sendSnap    = "sendMsgSnap"
)

var (
	ConnReadTimeout  = DefaultConnReadTimeout
	ConnWriteTimeout = DefaultConnWriteTimeout
)

type Peer interface {
	// 发送单个消息，该方法是非阻塞的，如采出现发送失败，则会将失败信息报告给底层的Raft接口
	send(m raftpb.Message)
	// 发送snap.Message，其他行为与上面的send（）方法类似
	sendSnap(m snap.Message)
	// 更新对应节点暴露的URL地址
	update(urls types.URLs)
	// ／将指定的连接与当前Peer绑定，Peer会将该连接作为Stream消息通道使用当Peer不再使用该连接时，会将该连接关闭
	attachOutgoingConn(conn *outgoingConn)
	activeSince() time.Time
	// 关闭当前Peer实例，会关闭底层的网络连接
	stop()
}

//结构体ra灿ttp.peer是Peer接口的具体实现
type peer struct {
	lg *zap.Logger
	localID types.ID
	id types.ID	// 该peer实例对应的节点的ID。
	r Raft	// Raft接口，在Raft接口实现的底层封装了etcd-raft模块。
	status *peerStatus
	// 每个节点可能提供了多个URL供其他节点访问，当其中一个访问失败时，我们应该可以尝试访问另一个。
	// 而urlPicker提供的主要功能就是在这些URL之间进行切换。
	picker *urlPicker
	msgAppV2Writer *streamWriter
	writer         *streamWriter	// 负责向S出am消息通道写入消息。
	pipeline       *pipeline	// Pipeline消息通道。
	snapSender     *snapshotSender // 负责发送快照数据。
	msgAppV2Reader *streamReader
	msgAppReader   *streamReader	// 负责从Stream消息通道读取消息。
	// 从Stream消息通道中读取到消息之后，会通过该通道将消息交给Raft接口，然后由它返回给底层etcd-raft模块进行处理。
	recvc chan raftpb.Message
	// 从Stream消息通道中读取到MsgProp类型的消息之后，会通过该通道将MsgProp消息交给Raft接口，
	// 然后由它返回给底层etcd-raft模块进行处理。
	propc chan raftpb.Message
	mu     sync.Mutex
	paused bool	// 是否暂停向对应节点发送消息。
	cancel context.CancelFunc // 取消对等方创建的go例程中的挂起工作。
	stopc  chan struct{}
}

// rafthttp.peer. startPeer（）方法中完成了初始化工作，同时也启动了关联的后台goroutine。
func startPeer(t *Transport, urls types.URLs, peerID types.ID, fs *stats.FollowerStats) *peer {
	if t.Logger != nil {
		t.Logger.Info("starting remote peer", zap.String("remote-peer-id", peerID.String()))
	}
	defer func() {
		if t.Logger != nil {
			t.Logger.Info("started remote peer", zap.String("remote-peer-id", peerID.String()))
		}
	}()

	status := newPeerStatus(t.Logger, t.ID, peerID)
	picker := newURLPicker(urls)	// 根据节点提供的URL创建urlPicker
	errorc := t.ErrorC
	r := t.Raft
	pipeline := &pipeline{	// 创建pipeine实例
		peerID:        peerID,
		tr:            t,
		picker:        picker,
		status:        status,
		followerStats: fs,
		raft:          r,
		errorc:        errorc,
	}
	pipeline.start()	// 启动pipeline

	p := &peer{	// 创建Peer实例
		lg:             t.Logger,
		localID:        t.ID,
		id:             peerID,
		r:              r,
		status:         status,
		picker:         picker,
		msgAppV2Writer: startStreamWriter(t.Logger, t.ID, peerID, status, fs, r),
		writer:         startStreamWriter(t.Logger, t.ID, peerID, status, fs, r),	// 创建并启动streamWriter
		pipeline:       pipeline,
		snapSender:     newSnapshotSender(t, picker, peerID, status),
		recvc:          make(chan raftpb.Message, recvBufSize),	// 创建recvc远远，注意缓冲区大小
		propc:          make(chan raftpb.Message, maxPendingProposals),	// 创建propc通道，注意缓冲区大小
		stopc:          make(chan struct{}),
	}

	ctx, cancel := context.WithCancel(context.Background())
	p.cancel = cancel
	// 启动单独的goroutine，它主妾负责将recvc通道中读取消息，该通道中的消息就是从对端节点发送过来的消息，
	// 然后将读取到的消息交给底层的Raft状态机进行处理
	go func() {
		for {
			select {
			case mm := <-p.recvc:	// 从recvc通道中获取连接上读取到的Message
				// 将Message交给底层Raft状态机处理，
				if err := r.Process(ctx, mm); err != nil {
					if t.Logger != nil {
						t.Logger.Warn("failed to process Raft message", zap.Error(err))
					}
				}
			case <-p.stopc:
				return
			}
		}
	}()

	// 在底层Raft状态机处理MsgProp类型的Message时，可能会阻塞，所以启动单独的goroutine来处理
	go func() {
		for {
			select {
			case mm := <-p.propc:	// 从propc通道中获取MsgProp类型的Message
				// 将Message交给底层Raft状态机处理，
				if err := r.Process(ctx, mm); err != nil {
					if t.Logger != nil {
						t.Logger.Warn("failed to process Raft message", zap.Error(err))
					}
				}
			case <-p.stopc:
				return
			}
		}
	}()

	// 创建并启动streamReader实例，它主要负责从Stream消息遥远上读取消息
	p.msgAppV2Reader = &streamReader{
		lg:     t.Logger,
		peerID: peerID,
		typ:    streamTypeMsgAppV2,
		tr:     t,
		picker: picker,
		status: status,
		recvc:  p.recvc,
		propc:  p.propc,
		rl:     rate.NewLimiter(t.DialRetryFrequency, 1),
	}
	p.msgAppReader = &streamReader{
		lg:     t.Logger,
		peerID: peerID,
		typ:    streamTypeMessage,
		tr:     t,
		picker: picker,
		status: status,
		recvc:  p.recvc,
		propc:  p.propc,
		rl:     rate.NewLimiter(t.DialRetryFrequency, 1),
	}

	p.msgAppV2Reader.start()
	p.msgAppReader.start()

	return p
}

func (p *peer) send(m raftpb.Message) {
	p.mu.Lock()
	paused := p.paused
	p.mu.Unlock()
	// 检测paused字段，是否暂停对指定节点发送消息
	if paused {
		return
	}
	// 根据消息的类型选择合适的消息通道，
	writec, name := p.pick(m)
	select {
	// 将Message写入writec通道中，等到发送
	case writec <- m:
	default:
		// 如采发送出现阻塞，则将信息报告给底层Raft状态机，这里也会根据消息类型选择合适的报告方法
		p.r.ReportUnreachable(m.To)
		if isMsgSnap(m) {
			p.r.ReportSnapshot(m.To, raft.SnapshotFailure)
		}
		if p.lg != nil {	// 记录日志
			p.lg.Warn(
				"dropped internal Raft message since sending buffer is full",
				zap.String("message-type", m.Type.String()),
				zap.String("local-member-id", p.localID.String()),
				zap.String("from", types.ID(m.From).String()),
				zap.String("remote-peer-id", p.id.String()),
				zap.String("remote-peer-name", name),
				zap.Bool("remote-peer-active", p.status.isActive()),
			)
		}
		sentFailures.WithLabelValues(types.ID(m.To).String()).Inc()
	}
}

func (p *peer) sendSnap(m snap.Message) {
	go p.snapSender.send(m)
}

func (p *peer) update(urls types.URLs) {
	p.picker.update(urls)
}

func (p *peer) attachOutgoingConn(conn *outgoingConn) {
	var ok bool
	switch conn.t {
	case streamTypeMsgAppV2:
		ok = p.msgAppV2Writer.attach(conn)
	case streamTypeMessage:
		ok = p.writer.attach(conn)
	default:
		if p.lg != nil {
			p.lg.Panic("unknown stream type", zap.String("type", conn.t.String()))
		}
	}
	if !ok {
		conn.Close()
	}
}

func (p *peer) activeSince() time.Time { return p.status.activeSince() }

// 暂停暂停对等方。对等方只需删除所有传入的
// 消息而不返回错误。
func (p *peer) Pause() {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.paused = true
	p.msgAppReader.pause()
	p.msgAppV2Reader.pause()
}

// 恢复将恢复暂停的对等方。
func (p *peer) Resume() {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.paused = false
	p.msgAppReader.resume()
	p.msgAppV2Reader.resume()
}

func (p *peer) stop() {
	if p.lg != nil {
		p.lg.Info("stopping remote peer", zap.String("remote-peer-id", p.id.String()))
	}

	defer func() {
		if p.lg != nil {
			p.lg.Info("stopped remote peer", zap.String("remote-peer-id", p.id.String()))
		}
	}()

	close(p.stopc)
	p.cancel()
	p.msgAppV2Writer.stop()
	p.writer.stop()
	p.pipeline.stop()
	p.snapSender.stop()
	p.msgAppV2Reader.stop()
	p.msgAppReader.stop()
}

// peer.pick（）方法中，会根据消息的类型选择合适的消息通道，并返回相应的通道供send()方法写入待发迭的消息
func (p *peer) pick(m raftpb.Message) (writec chan<- raftpb.Message, picked string) {
	var ok bool
	// 如果是MsgSnap类型的消息，则返回Pipeline消息迢迢对反的Channel，
	// 否则返回Stream消息通道对应的Channel，
	// 如果Stream消息远远不可用，则使用Pipeline消息远远发送所有类型的消息
	if isMsgSnap(m) {
		return p.pipeline.msgc, pipelineMsg
	} else if writec, ok = p.msgAppV2Writer.writec(); ok && isMsgApp(m) {
		return writec, streamAppV2
	} else if writec, ok = p.writer.writec(); ok {
		return writec, streamMsg
	}
	return p.pipeline.msgc, pipelineMsg
}

func isMsgApp(m raftpb.Message) bool { return m.Type == raftpb.MsgApp }

func isMsgSnap(m raftpb.Message) bool { return m.Type == raftpb.MsgSnap }
